package com.example.biyesheji.controller;

import com.example.biyesheji.mapper.ProductMapper;
import com.example.biyesheji.pojo.ProductSales;
import lombok.SneakyThrows;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 销售统计
 *
 * @author xwh
 * @date 2023/02/08
 */
@RestController
@CrossOrigin
@RequestMapping("/product")
public class SaleStaticticsController implements Serializable {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private JavaSparkContext sc;

/*
    @Autowired
    private ScheduledExecutorService scheduledExecutorService;

    //总数据列表
    List<Float> floatList = new ArrayList<>();

    //sse
    SseEmitter sseEmitter = new SseEmitter();

    //计数器
    Long i = 0l;

    @GetMapping("/sale")
    public SseEmitter getSale(){

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                i++;
                List<Float> oneColumn = productMapper.getOnePrice(i);

                floatList.add(oneColumn.get(0));

                //RDD计算总值
                //内存中加载数据
                JavaRDD<Float> rdd = sc.parallelize(floatList);

                Float reduce = rdd.reduce((Function2<Float, Float, Float>) (a, b) -> a + b);

                //清空之前数据，保留计算值
                floatList.clear();
                floatList.add(reduce);

                //rdd消除
                //rdd.unpersist();

                sseEmitter.send(floatList);
            }
        },10,5, TimeUnit.SECONDS);
        return sseEmitter;

    }

 */
    @GetMapping("/sale")
    public Float getSale(){
        List<Float> toalPrice = productMapper.getToalPrice();

        //RDD计算总值
        //内存中加载数据
        JavaRDD<Float> rdd = sc.parallelize(toalPrice);

        Float reduce = rdd.reduce((Function2<Float, Float, Float>) (a, b) -> a + b);
        rdd.unpersist();
        return reduce;
    }

}
